package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators
import org.apache.spark.ml.linalg.SparseVector

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.clustering._

/**
  * 本代码执行报错，可能是由于代码结构不适合zipWithIndex方法。真正使用的话，需要将代码重构。
  * Created by Administrator on 2017/4/27 0027.
  * 在路透社电报新闻上运行LDA算法
  * 其实，构建词典可以用sc.textFile("data/rcorpus").flatMap(_.split(" "))   ，本程序中没有使用这种方法。
  */
object TestLDA {


  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)

    def bagsFromDocumentPerLine(filename:String)=
      //所有内容经过split()操作后会转化为传统的Scala集合（而不是RDD）。因此，后续采用的是groupBy()这种Scala集合上的
      //操作方法，而不是RDD上的groupByKey(),尽管它们功能相似。
      sc.textFile(filename).map(_.split(" ").filter(x => x.length > 5 && x.toLowerCase != "reuter")
        .map(_.toLowerCase)
        .groupBy(x => x)
        .toList
        .map(x=>(x._1, x._2.size)))

    //读取文件（rcorpus），文件里面的每一行形成了RDD的一个元素，这个元素是一个List（词袋）。
    //rddBags是个rdd，其中每个元素是这样的形式：
    //List((application,1), (network,2), (assets,1), (effort,1), (commerce,2), (create,1), (harris,1), (county,1), (bankhouston,1), (bancshares,1), (deposits,1), (largest,1), (billion,2), (banking,1), (comptroller,1), (currency,1), (having,1))
    val rddBags:RDD[List[Tuple2[String,Int]]] = bagsFromDocumentPerLine("data/rcorpus")


    //vocab是全局Array，针对所有文章中曾经出现过的词；这个array的每个元素是一个tuple2，第一个是单词，第二个是单词的index（索引）。
    /*
      (skanska,0)
      (francs,1)
      (constraints,2)
      (drummond,3)
      (brewery,4)
      (secure,5)
      (sophistication,6)
      ...
      ...
      ...
     */
    val vocab: Array[Tuple2[String, Long]] =
      rddBags.flatMap(x => x)
              .reduceByKey(_ + _)
              .map(_._1)
              .zipWithIndex.cache()
              .collect

    def codeBags(rddBags:RDD[List[Tuple2[String,Int]]]) =
      rddBags.map( x => (x ++ vocab).groupBy(_._1)
                        .filter(_._2.size > 1)
                        .map(x =>(x._2(1)._2.asInstanceOf[Long].toInt,x._2(0)._2.asInstanceOf[Int].toDouble)).toList)//下一行的map()函数中的x格式如下：routine -> List((routine,1)。实际上传入map操作之前，程序是返回了一个map类型的数据：Map(routine -> List((routine,1), (routine,1529)), quality -> List((quality,1), (quality,3393)), cumulative -> List((cumulative,1),(cumulative,1648)),.........)
            .zipWithIndex().cache().map(x => (x._2, new SparseVector(vocab.size, x._1.map(_._1).toArray,x._1.map(_._2).toArray).asInstanceOf[Vector]))//zipWithIndex并没有改变每个词的编号。在这里，zipWithIndex是给了每个文档一个编号。


    val list_codebags = codeBags(rddBags).collect
    for(codebag <- list_codebags){
      codebag
    }
//    val model = new LDA().setK(5).run(codeBags(rddBags))
//    val list_topics = model.describeTopics(6).map(_._1.map(vocab(_)._1))
//    for(topic <- list_topics){
//      println(topic)
//    }
  }
}
